package cn.learn.server;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.*;
import java.util.concurrent.LinkedBlockingQueue;

public class NIOServerCnxnFactory extends ServerCnxnFactory {

    ServerSocketChannel ss;
    private final Set<SelectorThread> selectorThreads = new HashSet<SelectorThread>();
    private volatile boolean stopped = true;
    private AcceptThread acceptThread;

    @Override
    void configure(InetSocketAddress addr) throws IOException {
        System.out.println(addr);
        selectorThreads.add(new SelectorThread(1));

        this.ss = ServerSocketChannel.open();
        ss.socket().setReuseAddress(true);
        ss.socket().bind(addr);
        ss.configureBlocking(false);
        // 新开一个线程
        this.acceptThread = new AcceptThread(ss, addr, selectorThreads);
    }

    @Override
    public void startup(ZooKeeperServer zkServer, boolean startServer) {
        start();
        setZooKeeperServer(zkServer);
        if (startServer) {
            // 加载数据
            zkServer.startdata();
            // 启动
            zkServer.startup();
        }
    }

    @Override
    public void addSession(long sessionId, NettyServerCnxn nettyServerCnxn) {

    }

    private void start() {
        stopped = false;
        for (SelectorThread thread : selectorThreads) {
            if (thread.getState() == Thread.State.NEW) {
                thread.start();
            }
        }
        // ensure thread is started once and only once
        if (acceptThread.getState() == Thread.State.NEW) {
            acceptThread.start();
        }
    }

    private abstract class AbstractSelectThread extends Thread {
        protected final Selector selector;

        public AbstractSelectThread(String name) throws IOException {
            this.selector = Selector.open();
        }

        public void wakeupSelector() {
            selector.wakeup();
        }
    }

    private class AcceptThread extends AbstractSelectThread {
        ServerSocketChannel acceptSocket;
        private final SelectionKey acceptKey;

        private final Collection<SelectorThread> selectorThreads;
        private Iterator<SelectorThread> selectorIterator;

        public AcceptThread(ServerSocketChannel ss, InetSocketAddress addr, Set<SelectorThread> selectorThreads) throws IOException {
            super("NIOServerCxnFactory.AcceptThread:" + addr);
            this.acceptSocket = ss;
            this.acceptKey = acceptSocket.register(selector, SelectionKey.OP_ACCEPT);
            this.selectorThreads = Collections.unmodifiableList(new ArrayList<SelectorThread>(selectorThreads));
            this.selectorIterator = this.selectorThreads.iterator();
        }

        @Override
        public void run() {
            try {
                while (!stopped && !acceptSocket.socket().isClosed()) {
                    try {
                        select();
                    } catch (Exception e) {

                    }
                }
            } finally {
            }
        }

        private void select() {
            try {
                selector.select();
                Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
                while (!stopped && selectedKeys.hasNext()) {
                    SelectionKey key = selectedKeys.next();
                    selectedKeys.remove();
                    if (!key.isValid()) {
                        continue;
                    }
                    if (key.isAcceptable()) {
                        if (!doAccept()) {
//                            pauseAccept(10);
                        }
                    }
                }
            } catch (Exception e) {

            }
        }

        private boolean doAccept() {
            boolean accepted = false;
            SocketChannel sc = null;
            try {
                sc = acceptSocket.accept();
                accepted = true;
                InetAddress ia = sc.socket().getInetAddress();

                int cnxncount = getClientCnxnCount(ia);

                sc.configureBlocking(false);
                if (!selectorIterator.hasNext()) {
                    selectorIterator = selectorThreads.iterator();
                }
                SelectorThread selectorThread = selectorIterator.next();

                boolean bool = selectorThread.addAcceptedConnection(sc);
            } catch (Exception e) {

            }
            return accepted;
        }
    }

    private int getClientCnxnCount(InetAddress cl){
        return 0;
    }

    public class SelectorThread extends AbstractSelectThread {
        private final int id;
        private final Queue<SocketChannel> acceptedQueue;
        private final Queue<SelectionKey> updateQueue;

        public SelectorThread(int id) throws IOException {
            super("NIOServerCxnFactory.SelectorThread-" + id);
            this.id = id;
            this.acceptedQueue = new LinkedBlockingQueue<SocketChannel>();
            this.updateQueue = new LinkedBlockingQueue<SelectionKey>();
        }

        public boolean addAcceptedConnection(SocketChannel accepted) {
            if (stopped || !acceptedQueue.offer(accepted)) {
                return false;
            }
            wakeupSelector();
            return true;
        }

        @Override
        public void run() {
            while (!stopped) {
                select();
                processAcceptedConnections();
            }
        }

        private void processAcceptedConnections() {
            SocketChannel accepted;
            while (!stopped && (accepted = acceptedQueue.poll()) != null) {
                SelectionKey key = null;
                try {
                    key = accepted.register(selector, SelectionKey.OP_READ);
                    NIOServerCnxn cnxn = createConnection(accepted, key, this);
                    key.attach(cnxn);
                    addCnxn(cnxn);
                }catch (IOException e) {

                }
            }
        }

        private void select() {
            try {
                selector.select();
                Set<SelectionKey> selected = selector.selectedKeys();
                ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>(selected);
                Collections.shuffle(selectedList);  // 随机排列
                Iterator<SelectionKey> selectedKeys = selectedList.iterator();
                while (!stopped && selectedKeys.hasNext()) {
                    SelectionKey key = selectedKeys.next();
                    selected.remove(key);
                    if (!key.isValid()) {
                        cleanupSelectionKey(key);
                        continue;
                    }
                    if (key.isReadable() || key.isWritable()) {
                        System.out.println("readable:" + key);
                        handleIO(key);
                    } else {
                        System.out.println("Unexpected ops in select {" + key.readyOps() + "}");
                    }
                }
            } catch (IOException e) {
                e.printStackTrace();
            }
        }

        private void handleIO(SelectionKey key) {
            // 包装一下key
            IOWorkRequest workRequest = new IOWorkRequest(this, key);
            NIOServerCnxn cnxn = (NIOServerCnxn) key.attachment();

            key.interestOps(0);
        }

        protected void cleanupSelectionKey(SelectionKey key) {
            if (key != null) {
                try {
                    key.cancel();
                } catch (Exception ex) {
                    System.out.println(ex);
                }
            }
        }
    }

    private void addCnxn(NIOServerCnxn cnxn) throws IOException {
//        InetAddress addr = cnxn.getSocketAddress();
    }

    private NIOServerCnxn createConnection(SocketChannel sock, SelectionKey sk, SelectorThread selectorThread) {
        return new NIOServerCnxn(zkServer, sock, sk, this, selectorThread);
    }

    private class IOWorkRequest {
        private final SelectorThread selectorThread;
        private final SelectionKey key;

        public IOWorkRequest(SelectorThread selectorThread, SelectionKey key) {
            this.selectorThread = selectorThread;
            this.key = key;
        }
    }
}
